package com.example.demo.config;

import com.alibaba.fastjson2.JSON;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.example.demo.bean.OrderStatus;
import com.example.demo.bean.OrderStatusChangeEvent;
import com.example.demo.bean.StateMachineState;
import com.example.demo.mapper.StateMachineStateMapper;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.messaging.MessageHeaders;
import org.springframework.statemachine.StateMachineContext;
import org.springframework.statemachine.StateMachinePersist;
import org.springframework.statemachine.kryo.MessageHeadersSerializer;
import org.springframework.statemachine.kryo.StateMachineContextSerializer;
import org.springframework.statemachine.kryo.UUIDSerializer;
import org.springframework.statemachine.persist.DefaultStateMachinePersister;
import org.springframework.statemachine.persist.RepositoryStateMachinePersist;
import org.springframework.statemachine.persist.StateMachinePersister;
import org.springframework.statemachine.redis.RedisStateMachineContextRepository;
import org.springframework.statemachine.redis.RedisStateMachinePersister;
import org.springframework.statemachine.support.DefaultStateMachineContext;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;

@Configuration
@Slf4j
public class Persist<E, S> {

    @Resource
    private StateMachineStateMapper mapper;

    /**
     * 持久化到内存map中
     *
     * @return
     */
    @Bean(name = "stateMachineMemPersister")
    public StateMachinePersister<OrderStatus, OrderStatusChangeEvent, Long> getPersister() {
        return new DefaultStateMachinePersister<OrderStatus, OrderStatusChangeEvent, Long>(new StateMachinePersist<OrderStatus, OrderStatusChangeEvent, Long>() {

            @Override
            public void write(StateMachineContext<OrderStatus, OrderStatusChangeEvent> stateMachineContext, Long orderId) throws Exception {
                byte[] machineData = serialize(stateMachineContext);
                StateMachineState stateMachineState = mapper.selectByOrderId(orderId);
                if (Objects.nonNull(stateMachineState)) {
                    mapper.deleteById(stateMachineState.getId());
                }
                StateMachineState stateBean = new StateMachineState();
                stateBean.setState(stateMachineContext.getState().name());
                stateBean.setOrderId(orderId);
                stateBean.setMachineJson(machineData);
                stateBean.setIsDelete(Boolean.FALSE);
                mapper.insert(stateBean);
            }

            @Override
            public StateMachineContext<OrderStatus, OrderStatusChangeEvent> read(Long orderId) throws Exception {
                StateMachineState stateMachineState = mapper.selectByOrderId(orderId);
                if (Objects.nonNull(stateMachineState)) {
                    byte[] machineJson = stateMachineState.getMachineJson();
                    return deserialize(machineJson);
                }
                return null;
            }
        });
    }
 
    @Resource
    private RedisConnectionFactory redisConnectionFactory;
    private static final ThreadLocal<Kryo> KRYO_THREAD_LOCAL = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        kryo.addDefaultSerializer(StateMachineContext.class, new StateMachineContextSerializer());
        kryo.addDefaultSerializer(MessageHeaders.class, new MessageHeadersSerializer());
        kryo.addDefaultSerializer(UUID.class, new UUIDSerializer());
        return kryo;
    });

    private byte[] serialize(StateMachineContext<OrderStatus, OrderStatusChangeEvent> context) {
        Kryo kryo = KRYO_THREAD_LOCAL.get();
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        Output output = new Output(out);
        kryo.writeObject(output, context);
        output.close();
        return out.toByteArray();
    }

    @SuppressWarnings("unchecked")
    private StateMachineContext<OrderStatus, OrderStatusChangeEvent> deserialize(byte[] data) {
        if (data == null || data.length == 0) {
            return null;
        }
        Kryo kryo = KRYO_THREAD_LOCAL.get();
        ByteArrayInputStream in = new ByteArrayInputStream(data);
        Input input = new Input(in);
        return kryo.readObject(input, StateMachineContext.class);
    }
}